package com.jzl.canal_consumer.util;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;
import com.jzl.util.LoggerUtil;
import org.apache.commons.lang.StringUtils;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.io.UnsupportedEncodingException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

import static com.jzl.canal_consumer.util.LogFormat.*;

@Component
public class MessageEntryUtil
{

    public void printSummary(Message message, long batchId, int size)
    {
        long memsize = 0;
        for (CanalEntry.Entry entry : message.getEntries())
        {
            memsize += entry.getHeader().getEventLength();
        }

        String startPosition = null;
        String endPosition = null;
        if (!CollectionUtils.isEmpty(message.getEntries()))
        {
            startPosition = buildPositionForDump(message.getEntries().get(0));
            endPosition = buildPositionForDump(message.getEntries().get(message.getEntries().size() - 1));
        }

        SimpleDateFormat format = new SimpleDateFormat(LogFormat.DATE_FORMAT);
        LoggerUtil.info(LogFormat.contextFormat, new Object[]{batchId, size, memsize, format.format(new Date()), startPosition,
                endPosition});
    }

    public String buildPositionForDump(CanalEntry.Entry entry)
    {
        long time = entry.getHeader().getExecuteTime();
        Date date = new Date(time);
        SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
        String position = entry.getHeader().getLogfileName() + ":" + entry.getHeader().getLogfileOffset() + ":"
                + entry.getHeader().getExecuteTime() + "(" + format.format(date) + ")";
        if (StringUtils.isNotEmpty(entry.getHeader().getGtid()))
        {
            position += " gtid(" + entry.getHeader().getGtid() + ")";
        }
        return position;
    }

    public void printEntry(List<CanalEntry.Entry> entrys)
    {
        for (CanalEntry.Entry entry : entrys)
        {
            long executeTime = entry.getHeader().getExecuteTime();
            long delayTime = new Date().getTime() - executeTime;
            Date date = new Date(entry.getHeader().getExecuteTime());
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND)
            {
                if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN)
                {
                    CanalEntry.TransactionBegin begin = null;
                    try
                    {
                        begin = CanalEntry.TransactionBegin.parseFrom(entry.getStoreValue());
                    } catch (InvalidProtocolBufferException e)
                    {
                        throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
                    }
                    // 打印事务头信息，执行的线程id，事务耗时
                    LoggerUtil.debug(LogFormat.transactionFormat,
                            new Object[]{entry.getHeader().getLogfileName(),
                                    String.valueOf(entry.getHeader().getLogfileOffset()),
                                    String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date),
                                    entry.getHeader().getGtid(), String.valueOf(delayTime)});
                    LoggerUtil.debug(" BEGIN ----> Thread id: {}", begin.getThreadId());
                    printXAInfo(begin.getPropsList());
                } else if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND)
                {
                    CanalEntry.TransactionEnd end = null;
                    try
                    {
                        end = CanalEntry.TransactionEnd.parseFrom(entry.getStoreValue());
                    } catch (InvalidProtocolBufferException e)
                    {
                        throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
                    }
                    // 打印事务提交信息，事务id
                    LoggerUtil.debug("----------------\n");
                    LoggerUtil.debug(" END ----> transaction id: {}", end.getTransactionId());
                    printXAInfo(end.getPropsList());
                    LoggerUtil.debug(LogFormat.transactionFormat,
                            new Object[]{entry.getHeader().getLogfileName(),
                                    String.valueOf(entry.getHeader().getLogfileOffset()),
                                    String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date),
                                    entry.getHeader().getGtid(), String.valueOf(delayTime)});
                }

                continue;
            }

            if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA)
            {
                CanalEntry.RowChange rowChage = null;
                try
                {
                    rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                } catch (Exception e)
                {
                    throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
                }

                CanalEntry.EventType eventType = rowChage.getEventType();

                LoggerUtil.info(LogFormat.rowFormat,
                        new Object[]{entry.getHeader().getLogfileName(),
                                String.valueOf(entry.getHeader().getLogfileOffset()), entry.getHeader().getSchemaName(),
                                entry.getHeader().getTableName(), eventType,
                                String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date),
                                entry.getHeader().getGtid(), String.valueOf(delayTime)});

                if (eventType == CanalEntry.EventType.QUERY || rowChage.getIsDdl())
                {
                    LoggerUtil.info(" sql ----> " + rowChage.getSql() + SEP);
                    continue;
                }

                printXAInfo(rowChage.getPropsList());
                for (CanalEntry.RowData rowData : rowChage.getRowDatasList())
                {
                    if (eventType == CanalEntry.EventType.DELETE)
                    {
                        printColumn(rowData.getBeforeColumnsList());
                    } else if (eventType == CanalEntry.EventType.INSERT)
                    {
                        printColumn(rowData.getAfterColumnsList());
                    } else
                    {
                        printColumn(rowData.getAfterColumnsList());
                    }
                }
            }
        }
    }

    public void printColumn(List<CanalEntry.Column> columns)
    {
        StringBuilder builder = new StringBuilder();
        builder.append(" ");
        for (CanalEntry.Column column : columns)
        {

            try
            {
                if (StringUtils.containsIgnoreCase(column.getMysqlType(), "BLOB")
                        || StringUtils.containsIgnoreCase(column.getMysqlType(), "BINARY"))
                {
                    // get value bytes
                    builder.append(column.getName() + " : "
                            + new String(column.getValue().getBytes("ISO-8859-1"), "UTF-8"));
                } else
                {
                    builder.append(column.getName() + " : " + column.getValue());
                }
            } catch (UnsupportedEncodingException e)
            {
            }
            builder.append("    type=" + column.getMysqlType());
            if (column.getUpdated())
            {
                builder.append("    update=" + column.getUpdated());
            }
            builder.append(SEP);

        }
        LoggerUtil.debug(builder.toString());
    }

    public void printXAInfo(List<CanalEntry.Pair> pairs)
    {
        if (pairs == null)
        {
            return;
        }

        String xaType = null;
        String xaXid = null;
        for (CanalEntry.Pair pair : pairs)
        {
            String key = pair.getKey();
            if (StringUtils.endsWithIgnoreCase(key, "XA_TYPE"))
            {
                xaType = pair.getValue();
            } else if (StringUtils.endsWithIgnoreCase(key, "XA_XID"))
            {
                xaXid = pair.getValue();
            }
        }

        if (xaType != null && xaXid != null)
        {
            LoggerUtil.debug(" ------> " + xaType + " " + xaXid);
        }
    }

    /**
     * 打印当前batch的摘要信息
     *
     * @param message
     * @param batchId
     */
    public void printBatch(Message message, long batchId)
    {

        List<CanalEntry.Entry> entries = message.getEntries();
        if (CollectionUtils.isEmpty(entries))
        {
            return;
        }

        long memSize = 0;
        for (CanalEntry.Entry entry : entries)
        {
            memSize += entry.getHeader().getEventLength();
        }
        int size = entries.size();
        String startPosition = buildPosition(entries.get(0));
        String endPosition = buildPosition(message.getEntries().get(size - 1));

        SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
        LoggerUtil.info(LogFormat.contextFormat, new Object[]{batchId, size, memSize, format.format(new Date()), startPosition, endPosition});
    }

    public String buildPosition(CanalEntry.Entry entry)
    {
        CanalEntry.Header header = entry.getHeader();
        long time = header.getExecuteTime();
        Date date = new Date(time);
        SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
        StringBuilder sb = new StringBuilder();
        sb.append(header.getLogfileName())
                .append(":")
                .append(header.getLogfileOffset())
                .append(":")
                .append(header.getExecuteTime())
                .append("(")
                .append(format.format(date))
                .append(")");
        return sb.toString();
    }

    /**
     * default，only logging information
     *
     * @param entry
     */
    public void transactionBegin(CanalEntry.Entry entry)
    {

        try
        {
            CanalEntry.TransactionBegin begin = CanalEntry.TransactionBegin.parseFrom(entry.getStoreValue());
            // 打印事务头信息，执行的线程id，事务耗时
            CanalEntry.Header header = entry.getHeader();
            long executeTime = header.getExecuteTime();
            long delayTime = new Date().getTime() - executeTime;
            LoggerUtil.debug(LogFormat.transactionFormat,
                    new Object[]{"begin", begin.getTransactionId(), header.getLogfileName(),
                            String.valueOf(header.getLogfileOffset()),
                            String.valueOf(header.getExecuteTime()), String.valueOf(delayTime)});
        } catch (Exception e)
        {
            LoggerUtil.error("parse event has an error , data:" + entry.toString(), e);
        }
    }

    public void transactionEnd(CanalEntry.Entry entry)
    {

        try
        {
            CanalEntry.TransactionEnd end = CanalEntry.TransactionEnd.parseFrom(entry.getStoreValue());
            // 打印事务提交信息，事务id
            CanalEntry.Header header = entry.getHeader();
            long executeTime = header.getExecuteTime();
            long delayTime = new Date().getTime() - executeTime;

            LoggerUtil.debug(LogFormat.transactionFormat,
                    new Object[]{"end", end.getTransactionId(), header.getLogfileName(),
                            String.valueOf(header.getLogfileOffset()),
                            String.valueOf(header.getExecuteTime()), String.valueOf(delayTime)});
        } catch (Exception e)
        {
            LoggerUtil.error("parse event has an error , data:" + entry.toString(), e);
        }
    }

    /**
     * 判断事件类型为DML 数据
     *
     * @param eventType
     * @return
     */
    public boolean isDML(CanalEntry.EventType eventType)
    {
        switch (eventType)
        {
            case INSERT:
            case UPDATE:
            case DELETE:
                return true;
            default:
                return false;
        }
    }

    /**
     * 处理 DDL数据
     *
     * @param header
     * @param eventType
     * @param sql
     */

    public void processDDL(CanalEntry.Header header, CanalEntry.EventType eventType, String sql)
    {

        String table = header.getSchemaName() + "." + header.getTableName();
        //对于DDL，直接执行，因为没有行变更数据
        switch (eventType)
        {
            case CREATE:
                LoggerUtil.warn("parse create table event, table: {}, sql: {}", table, sql);

                return;
            case ALTER:
                LoggerUtil.warn("parse alter table event, table: {}, sql: {}", table, sql);

                return;
            case TRUNCATE:
                LoggerUtil.warn("parse truncate table event, table: {}, sql: {}", table, sql);
                return;
            case ERASE:
            case QUERY:
                LoggerUtil.warn("parse event : {}, sql: {} . ignored!", eventType.name(), sql);
                return;
            case RENAME:
                LoggerUtil.warn("parse rename table event, table: {}, sql: {}", table, sql);
                return;
            case CINDEX:
                LoggerUtil.warn("parse create index event, table: {}, sql: {}", table, sql);
                return;
            case DINDEX:
                LoggerUtil.warn("parse delete index event, table: {}, sql: {}", table, sql);
                return;
            default:
                LoggerUtil.warn("parse unknown event: {}, table: {}, sql: {}", eventType.name(), table, sql);
                break;
        }
    }

    /**
     * 强烈建议捕获异常，非上述已列出的其他操作，非核心
     * 除了“insert”、“update”、“delete”操作之外的，其他类型的操作.
     * 默认实现为“无操作”
     *
     * @param header 可以从header中获得schema、table的名称
     * @param sql
     */
    public void whenOthers(CanalEntry.Header header, String sql)
    {
        String schema = header.getSchemaName();
        String table = header.getTableName();
        LoggerUtil.error("ignore event,schema: {},table: {},SQL: {}", schema, table, sql);
    }

    public enum ExceptionStrategy
    {
        RETRY(1), IGNORE(2);
        public int code;

        ExceptionStrategy(int code)
        {
            this.code = code;
        }

        public static ExceptionStrategy codeOf(Integer code)
        {
            if (code != null)
            {
                for (ExceptionStrategy e : ExceptionStrategy.values())
                {
                    if (e.code == code)
                    {
                        return e;
                    }
                }
            }
            return null;
        }
    }
}
